package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.dws.reservecharge.ReserveChargeData;
import cn.gwm.flink.streaming.dws.reservecharge.test.CustomJsonSource;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

@Slf4j
public class AliyunKafkaTestTask extends BaseTask {
    private static final String DASK_NAME = "aliyunTestTask";
    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        StreamExecutionEnvironment env = getEnv(DASK_NAME);
        env.enableCheckpointing(120 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(240 * 1000L);
        env.setParallelism(3);

        DataStream<ReserveChargeData> sourceDataStream = env.addSource(new CustomJsonSource())
                .map(str -> JSONUtil.toBean(str, ReserveChargeData.class));

//        defaultSinkToKafka(process, ConfigLoader.get("kafka.dws_receive.topic"), "id", null, null);
        defaultSinkToKafka(sourceDataStream, "test", "id", null, null);
        sourceDataStream.print();
        env.execute(DASK_NAME);


    }
}
